# !/usr/bin/env python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from pyspark.sql.functions import isnull, when, count, col
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# https://www.kaggle.com/uciml/pima-indians-diabetes-database?ref=hackernoon.com
'''
synapse.ml
import pyspark
spark = pyspark.sql.SparkSession.builder.appName("MyApp") \
            .config("spark.jars.packages", "com.microsoft.azure:synapseml_2.12:1.0.4") \
            .getOrCreate()
import synapse.ml
'''
'''
Pregnancies：怀孕次数
Glucose：2小时内口服葡萄糖耐量试验的血糖浓度
BloodPressure：舒张压(mm Hg)
SkinThickness：三头肌皮肤褶皱厚度(mm)
Insulin：2小时血清胰岛素(mu U/ml)
BMI：身体质量指数(体重单位kg/(身高单位m)²)
diabespedigreefunction：糖尿病谱系功能
Age：年龄(年)
Outcome：类变量(0或1)
输入变量： 葡萄糖、血压、BMI、年龄、怀孕、胰岛素、皮肤厚度、糖尿病谱系函数。
输出变量： 结果。
'''
'''
在分类(classification)问题的模型评估中，常用的评测指标有以下7个：
准确率(accuracy)
精确率(precision)
召回率(recall)
F1-Score
ROC曲线
P-R曲线
AUC面积
https://blog.csdn.net/michael_f2008/article/details/104938824
'''


spark = SparkSession.builder.appName("synapse learn")\
    .config("spark.jars.packages", "com.microsoft.azure:synapseml_2.12:1.0.4").master("local[*]").getOrCreate()
try:
    from synapse.ml.lightgbm import LightGBMClassifier
    from synapse.ml.lightgbm import LightGBMRegressor
except:
    from synapse.ml.LightGBMClassifier import LightGBMClassifier
    from synapse.ml.LightGBMRegressor import LightGBMRegressor


def get_classifier(name, label='Outcome'):
    if name == 'LightGBM':
        # LightGBM（Light Gradient Boosting Machine）是一个实现GBDT算法的框架
        return LightGBMClassifier(labelCol=label, featuresCol='features', maxDepth=3)
    return None


if __name__ == '__main__':
    # spark = SparkSession.builder.appName("classifier_learn").master("local[*]").getOrCreate()

    df = spark.read.csv('../data/diabetes.csv', header=True, inferSchema=True)
    df.printSchema()

    # 缺失数据处理
    df.select([count(when(isnull(c), c)).alias(c) for c in df.columns]).show()

    # 不必要列丢弃
    dataset = df.drop('SkinThickness').drop('Insulin').drop('DiabetesPedigreeFunction').drop('Pregnancies')
    dataset.show()

    # 特征转换向量
    # 用VectorAssembler合并所有特性
    required_features = ['Glucose', 'BloodPressure', 'BMI', 'Age']
    assembler = VectorAssembler(inputCols=required_features, outputCol='features')
    transformed_data = assembler.transform(dataset)
    transformed_data.show()

    # 将数据随机分成训练集和测试集，并设置可重复性的种子。
    (training_data, test_data) = transformed_data.randomSplit([0.8, 0.2], seed=2020)
    print("训练数据集总数: " + str(training_data.count()))
    print("测试数据集总数: " + str(test_data.count()))

    class_arr = ["LightGBM"]
    for name in class_arr:
        rf = get_classifier(name)
        model = rf.fit(training_data)
        rf_predictions = model.transform(test_data)
        print(f"==================={name} prediction=====================")
        rf_predictions.show()
        #  评估随机森林分类器模型
        multi_evaluator = MulticlassClassificationEvaluator(labelCol='Outcome', metricName='accuracy')
        print(name + ' classifier Accuracy:', multi_evaluator.evaluate(rf_predictions))
    spark.stop()
